import config from './config.js'
import pack from './package.json'
import redis from 'redis'
import { createServer } from "http";
import { Server } from "socket.io";

// 输出端口地址
console.log("webredis start on:%d", config.PORT)
// 输出redis 地址
console.log("redis localtion:%s", config.REDIS_URL)

// 每一个websocket代表一个浏览器连接，使用socket.id为key保存频道订阅数据和socket实例
// 保存所有连接的socket
// socket.id=>socket
const WEB_SOCKETS = new Map()
// websocket订阅的频道集合
// socket.id=>Set<String>
const SUBSCRIBES = new Map()

// 1 连接redis
var redisClientError;
// 公用redis client实例
var redisClient = redis.createClient(config.REDIS_URL)
redisClient.on('error', function (error) {
  redisClientError = error;
  console.log(error);
});
var redisSubError;
// 只用于消息订阅subscribe的redis客户端,需要独占链接,全局变量
var redisSub = redis.createClient(config.REDIS_URL)
  .on("subscribe", function (channel, count) {
    /** 命令输出频道订阅日志 */
    console.log(`channel [${channel}] subscriber num: ${count}`);
  })
  .on('message', function (channel, message) {
    /** 
     * 分发订阅频道的收到的消息到订阅频道的连接
     * 遍历每个socket订阅的频道，检查当前消息的频道是在该socket订阅的频道名集合中，
     * 如果是就向该socket推送消息
     */
    console.log("recieve mq: channel=%s, message=%s", channel, message)
    for (var [socketid, channels] of SUBSCRIBES) {
      // 如果socket订阅了消息频道则推送消息    
      if (channels.has(channel)) {
        WEB_SOCKETS.get(socketid).emit(
          'message',
          {
            channel: channel,
            data: message
          })
      }
    }
  })
redisSub.on('error', function (error) {
  redisSubError = error;
  console.log(error);
});

// 2 创建http server
var httpServer = createServer(function (req, res) {
  res.writeHead(200, {
    'Content-Type': 'text/plain'
  });
  // 显示项目名称及版本号
  res.end('mq server: ' + pack.name + ' ' + pack.version + ' ' + pack.repository.url + ' \nlisten on ' + config.PORT);
})

// 3 创建ws server
var io = new Server(httpServer)
io.on('connection', function (socket) {
  console.log('$%s connect', socket.id)
  // 将socket对象保存到全局映射表中
  WEB_SOCKETS.set(socket.id, socket)
  // 初始化订阅频道表
  SUBSCRIBES.set(socket.id, new Set())
  /** 频道订阅  */
  socket.on('subscribe', function (channels) {
    var ack = {
      success: false,
      request: 'subscribe',
      reply: null,
      respone: 'OK'
    }
    try {
      if (typeof channels === 'string') {
        channels = [channels]
      }
      if (!(channels instanceof Array)) {
        ack.respone = 'channels is NOT ARRAY'
        socket.emit('ack', ack)
        return
      } else if (redisSub.ready === false || redisSub.stream.writable === false) {
        ack.respone = redisSubError.message
        socket.emit('ack', ack)
      } else if (channels.length > 0) {
        console.log("$%s subscribe channel: %s", socket.id, channels)
        redisSub.subscribe(channels, (err, res) => {
          if (err) {
            ack.respone = err.message
          } else {
            ack.success = true
            ack.respone = '订阅成功'
            ack.reply = res
            saveChannels(socket, channels);
            printChannels(channels)
            socket.emit('ack', ack)
          }
        })
      } else {
        ack.respone = '订阅的频道列表为空'
        socket.emit('ack', ack)
      }
    } catch (error) {
      ack.respone = err.message
      socket.emit('ack', ack)
    }
  })
  /** 取消订阅 */
  socket.on('unsubscribe', function (channels) {
    var ack = {
      success: false,
      request: 'unsubscribe',
      reply: null,
      respone: 'OK'
    }
    try {
      if (typeof channels === 'string') {
        channels = [channels]
      }
      if (!(channels instanceof Array)) {
        ack.respone = 'channels is NOT ARRAY'
        socket.emit('ack', ack)
      } else if (redisSub.ready === false || redisSub.stream.writable === false) {
        ack.respone = redisSubError.message
        socket.emit('ack', ack)
      } else if (channels.length > 0) {
        console.log("$%s unsubscribe channel: %s", socket.id, channels)
        redisClient.unsubscribe(channels, (err, res) => {
          if (err) {
            ack.respone = error.message
          } else {
            ack.success = true
            ack.respone = '取消订阅成功'
            ack.reply = res
            removeChannels(socket, channels);
          }
          socket.emit('ack', ack)
        })

      } else {
        ack.respone = '取消订阅的频道列表为空'
        socket.emit('ack', ack)
      }
    } catch (error) {
      ack.respone = error.message
      socket.emit('ack', ack)
    }
  })

  /** 发布消息 */
  socket.on('publish', (channel, message) => {
    var ack = {
      success: false,
      request: 'publish',
      reply: null,
      respone: 'OK'
    }
    try {
      if(!channel || !message){
        ack.respone = 'NULL INPUT (channel or message)'
        socket.emit('ack', ack)
      }else if (redisSub.ready === false || redisSub.stream.writable === false) {
        ack.respone = redisSubError.message
        socket.emit('ack', ack)
      } else {
        console.log("$%s publish mq: channel=%s, message=%s", socket.id, channel, message)
        redisClient.publish(channel, message, (err, res) => {
          if (err) {
            ack.respone = error.message
          } else {
            ack.success = true
            ack.respone = channel + '发布成功'
            ack.reply = res
          }
          socket.emit('ack', ack)
        });

      }
    } catch (error) {
      ack.respone = error.message
      socket.emit('ack', ack)
    }

  });

  /** 关闭websocket连接 */
  socket.on('disconnect', function (reason) {
    console.log('%s disconnect %s', socket.id, reason)
    WEB_SOCKETS.delete(socket.id)
    SUBSCRIBES.delete(socket.id)
  })
});


// 删除socket订阅的频道的记录
function removeChannels(socket, channels) {
  var set = SUBSCRIBES.get(socket.id);
  channels.forEach(element => {
    set.delete(element);
  });
}

// 记录每个socket订阅的频道
function saveChannels(socket, channels) {
  var set = SUBSCRIBES.get(socket.id);
  if(set){
	  channels.forEach(element => {
		set.add(element);
	  });
  }else{
	console.log("ERROR:INVALID socket!!! $%s", socket.id)
  }
}

// 打印频道订阅信息
function printChannels(channels){
    for (var channel of channels){
      var sids = []
      for (var [socketid, channels] of SUBSCRIBES) {
        // 如果socket订阅了消息频道则推送消息
        if (channels.has(channel)) {
          sids.push(socketid)
        }
      }
      console.log('channel [%s] subscriber: %s', channel, JSON.stringify(sids))
    }
}

// 4 启动http server
httpServer.listen(config.PORT);